package com.jy.rhin.application.infra;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.parser.Feature;
import com.jy.rhin.application.infra.assembler.PublishMessageAssembler;
import com.jy.rhin.application.infra.assembler.SubscribeAndFilterAssembler;
import com.jy.rhin.config.RhinRabbitConfig;
import com.jy.rhin.domain.exception.RhinInfraErrorType;
import com.jy.rhin.domain.notification.participant.ParticipantDomainService;
import com.jy.rhin.domain.notification.participant.entity.EnumParticipantRole;
import com.jy.rhin.domain.notification.participant.entity.ParticipantDO;
import com.jy.rhin.domain.notification.publish.PublishMessageDomainService;
import com.jy.rhin.domain.notification.publish.entity.PublishMessageDO;
import com.jy.rhin.domain.notification.publish.entity.SubscribeAndFilterVO;
import com.jy.rhin.domain.notification.receive.ReceiveMessageDomainService;
import com.jy.rhin.domain.notification.receive.entity.ReceiveMessageDO;
import com.jy.rhin.domain.notification.subscribe.SubscribeDomainService;
import com.jy.rhin.domain.notification.subscribe.entity.SubscribeDO;
import com.jy.rhin.domain.notification.subscribe.entity.vo.SubscribeTopicFilterVO;
import com.jy.rhin.domain.notification.topic.TopicDomainService;
import com.jy.rhin.domain.notification.topic.entity.TopicDO;
import com.jy.rhin.domain.notification.topic.entity.TopicQueryFilterVO;
import com.jy.rhin.domain.notification.topic.entity.TopicQueryVO;
import com.jy.rhin.facade.infra.controller.notification.assembler.SubscribeTopicAssembler;
import com.jy.rhin.infra.feign.QueryTopicConfigurationParametersService;
import com.jy.rhin.infra.repo.notfication.participant.query.ParticipantQueryHandler;
import com.jy.rhin.infra.repo.notfication.publish.query.PublishMessageQueryHandler;
import com.jy.rhin.infra.repo.notfication.topic.query.TopicQueryHandler;
import com.jy.rhin.infra.uitl.Asserts;
import com.jy.rhin.sdk.dto.notification.*;
import com.jy.rhin.sdk.dto.notification.req.CancelSubscribeDto;
import com.jy.rhin.sdk.dto.notification.req.IdDto;
import com.jy.rhin.sdk.dto.notification.resp.PublishMessageDetailDto;
import com.jy.rhin.sdk.dto.notification.resp.PublishMessageDto;
import com.jy.rhin.sdk.dto.notification.resp.SubscribeAndFilterDto;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author WeiQ
 * @date 2022/6/28 10:08
 * 消息总线服务
 */
@Slf4j
@Service
public class NotificationApplicationService {

    @Value("${spring.application.name}")
    private String serviceName;
    @Resource
    RabbitTemplate rabbitTemplate;
    @Resource
    QueryTopicConfigurationParametersService queryTopicConfigurationParametersService;
    @Resource
    RhinRabbitConfig.RhinRabbitConfigurationProperties rhinRabbitProperties;
    @Resource
    TopicDomainService topicDomainService;
    @Resource
    SubscribeDomainService subscribeDomainService;
    @Resource
    PublishMessageDomainService publishMessageDomainService;
    @Resource
    private ReceiveMessageDomainService receiveDomainService;
    @Resource
    private ParticipantDomainService participantDomainService;
    @Resource
    TopicQueryHandler topicQueryHandler;
    @Resource
    ParticipantQueryHandler participantQueryHandler;
    @Resource
    PublishMessageQueryHandler publishMessageQueryHandler;
    private static final String TOPIC_EXP = "topicExp";

    /**
     * 创建一个主题
     * 发布者必须存在
     *
     * @param topicDO
     * @return
     */
    public Long create(TopicDO topicDO) {
        //发布者验证
        ParticipantDO participantDO = participantQueryHandler.queryByCodeAndRole(topicDO.getPublisherCode(), EnumParticipantRole.PUBLISHER);
        Asserts.nonNull(participantDO, RhinInfraErrorType.PARTICIPANT_NOT_FOUND);
        return topicDomainService.create(topicDO);
    }

    /**
     * 校验消息内容topicExp
     *
     * @param message
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public Long receive(String message, String messageId, String messageSender, Date messageTime) {
        JSONObject messageContent = JSON.parseObject(message, Feature.IgnoreAutoType);
        String topicExp = messageContent.getString(TOPIC_EXP);
        TopicDO topic = Optional.ofNullable(topicDomainService.queryByTopicExp(messageContent.getString(TOPIC_EXP)))
                .orElseGet(() -> queryAndAddTopicConfigurationParameters(messageSender, topicExp));
        Asserts.nonNull(topic, RhinInfraErrorType.RECEIVE_TOPIC_NOT_FOUND);
        ReceiveMessageDO receive = new ReceiveMessageDO();
        receive.receiveMessage(messageId, topic.getId(), messageContent.toJSONString(), messageSender, messageTime);
        Long receiveId = receiveDomainService.receive(receive);
        List<SubscribeAndFilterDto> subscribeAndFilterList = publishMessageQueryHandler.querySubscribeAndFilterListByTopicId(topic.getId());
        //没有订阅者直接返回
        if (CollectionUtils.isEmpty(subscribeAndFilterList)) {
            return receiveId;
        }
        //将dto转换为domain可以使用的vo对象
        List<SubscribeAndFilterVO> dataList = SubscribeAndFilterAssembler.toSubscribeAndFilterVo(subscribeAndFilterList);
        List<PublishMessageDO> records = publishMessageDomainService.buildPublishMessage(dataList, receiveId, messageContent.toJSONString());
        Map<String, SubscribeAndFilterDto> subscribeAndFilterMap = subscribeAndFilterList.stream().collect(Collectors.toMap(SubscribeAndFilterDto::getSubscriberCode, dto -> dto, (t, t2) -> t));
        //过滤完存在符合条件的订阅者发送消息记录
        if (!CollectionUtils.isEmpty(records)) {
            sendMessage(PublishMessageAssembler.toDto(records, subscribeAndFilterMap, messageContent, messageId));
        }
        return receiveId;
    }

    /**
     * 查询以及将主题配置信息入库
     *
     * @param messageSender
     * @param topicExp
     * @return
     */
    private TopicDO queryAndAddTopicConfigurationParameters(String messageSender, String topicExp) {
        ParticipantDO participant = queryParticipant(messageSender);
        if (Objects.isNull(participant)) {
            log.error("未查询到主题且主题发布者不存在，忽略：{}------{}", topicExp, messageSender);
            return null;
        }
        TopicDO topic = queryTopicConfigurationParametersService.queryTopicSettings(participant.getParticipantUrl(), topicExp);
        if (Objects.isNull(topic)) {
            log.error("未获取主题配置信息，忽略！");
            return null;
        }
        topic.setPublisherCode(participant.getParticipantCode());
        create(topic);
        return topic;
    }

    /**
     * 根据消息发送者查询发布者
     * 无-则直接忽略 不做任何处理
     * 有-则根据服务地址获取主题配置信息存库 接着走业务流程
     *
     * @param messageSender
     */
    private ParticipantDO queryParticipant(String messageSender) {
        return participantQueryHandler.queryByCodeAndRole(messageSender, EnumParticipantRole.PUBLISHER);
    }

    /**
     * 通知消息详情获取
     *
     * @param req
     * @return
     */
    public PublishMessageDetailDto queryPublishMessageDetail(IdDto req) {
        return publishMessageQueryHandler.queryPublishMessageDetail(req.getId());
    }

    /**
     * 订阅主题
     *
     * @param dto
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public Long subscribe(SubscribeTopicDto dto) {
        SubscribeDO subscribeDO = subscribeCheck(dto);
        return subscribeDomainService.subscribe(subscribeDO);
    }

    /**
     * 订阅主题相关信息校验
     * @param dto
     * @return
     */
    private SubscribeDO subscribeCheck(SubscribeTopicDto dto) {
        SubscribeDO subscribeDO = new SubscribeDO(dto.getOperator());
        subscribeDO.subscribe(dto.getId(), dto.getSubscriberCode(), dto.getTopicExp(), dto.getQueryExp(), dto.getOperator());
        //查询函数存在时才能保存过滤条件
        if (!StringUtils.isEmpty(subscribeDO.getQueryExp())) {
            subscribeDO.setFilterList(SubscribeTopicAssembler.toFilterVO(dto.getFilters()));
        }
        //订阅者校验
        subscribeCheckSubscriber(subscribeDO.getSubscriberCode());
        //订阅绑定的主题和查询函数、过滤条件校验
        subscribeCheckTopic(subscribeDO);
        return subscribeDO;
    }

    /**
     * 取消订阅
     *
     * @param dto
     * @return
     */
    public Boolean cancelSubscribe(CancelSubscribeDto dto) {
        return subscribeDomainService.cancelSubscribe(dto.getId(), dto.getOperator());
    }
    
    /**
     * 暂停订阅
     *
     * @param dto
     * @return
     */
    public Boolean pauseSubscribe(CancelSubscribeDto dto) {
        return subscribeDomainService.pauseSubscribe(dto.getId(), dto.getOperator());
    }
    
    /**
     * 恢复订阅
     *
     * @param dto
     * @return
     */
    public Boolean resumeSubscribe(CancelSubscribeDto dto) {
        return subscribeDomainService.resumeSubscribe(dto.getId(), dto.getOperator());
    }


    /**
     * 修改、添加发布者
     *
     * @param dto
     * @return
     */
    public Long editPublisher(EditPublisherDto dto) {
        ParticipantDO participant = new ParticipantDO(dto.getId(), dto.getOperator());
        participant.publisher(dto.getPublisherName(), dto.getPublisherCode(), dto.getPublisherUrl(), dto.getPublisherDesc(), dto.getOperator());
        return participantDomainService.saveParticipant(participant);
    }

    /**
     * 修改、添加订阅者
     *
     * @param dto
     * @return
     */
    public Long editSubscriber(EditSubscriberDto dto) {
        ParticipantDO participant = new ParticipantDO(dto.getId(), dto.getOperator());
        participant.subscriber(dto.getSubscriberName(), dto.getSubscriberCode(), dto.getInteractionType(), dto.getSubscriberUrl(), dto.getSubscriberDesc(), dto.getOperator());
        return participantDomainService.saveParticipant(participant);
    }

    /**
     * 发布消息，记录发送操作结果
     *
     * @param subscribeId
     * @param publishResult
     * @param failReason
     */
    public void publishMessage(Long subscribeId, boolean publishResult, String failReason) {
        publishMessageDomainService.publishMessage(subscribeId, publishResult, failReason);
    }

    /**
     * 业务服务订阅主题
     *
     * @param dto
     * @return
     */
    @Transactional(rollbackFor = Exception.class)
    public Long businessSubscribeTopic(SubscriberAndSubscribeTopicDto dto) {
        BusinessSubscriberDto subscriber = dto.getSubscriber();
        BusinessSubscribeTopicDto subscriberTopic = dto.getSubscriberTopic();
        ParticipantDO participant = new ParticipantDO(null, dto.getOperator());
        participant.subscriber(subscriber.getSubscriberName(), subscriber.getSubscriberCode(), subscriber.getInteractionType(), subscriber.getSubscriberUrl(), subscriber.getSubscriberDesc(), dto.getOperator());
        //保存订阅者信息
        participantDomainService.addParticipant(participant);
        SubscribeTopicDto subscribeTopicDto = SubscribeTopicAssembler.toSubscribeTopicDto(subscriberTopic,
                subscriber.getSubscriberCode(),
                dto.getOperator());
        //保存主题订阅信息
        SubscribeDO subscribeDO = subscribeCheck(subscribeTopicDto);
        return subscribeDomainService.subscribe(subscribeDO);
    }

    /**
     * 订阅绑定的主题校验
     *
     * @param subscribeDO
     */
    private void subscribeCheckTopic(SubscribeDO subscribeDO) {
        String topicExp = subscribeDO.getTopicExp();
        String queryExp = subscribeDO.getQueryExp();
        if (!StringUtils.isEmpty(queryExp)) {
            subscribeCheckFilter(topicExp, queryExp, subscribeDO.getFilterList());
        } else {
            //订阅主题是否存在
            TopicDO topicDO = topicQueryHandler.queryTopicByExp(topicExp);
            Asserts.nonNull(topicDO, RhinInfraErrorType.SUBSCRIBE_TOPIC_IS_NULL);
        }
    }

    /**
     * 订阅绑定的查询函数、过滤条件校验
     *
     * @param topicExp
     * @param queryExp
     * @param filterList
     */
    private void subscribeCheckFilter(String topicExp, String queryExp, List<SubscribeTopicFilterVO> filterList) {
        TopicQueryVO topicQueryVo = topicQueryHandler.queryTopicQueryByExp(topicExp, queryExp);
        //查询函数存在时过滤条件不能为空
        Asserts.notEmpty(filterList, RhinInfraErrorType.SUBSCRIBE_TOPIC_FILTER_NOT_NULL);
        //预存查询函数是否存在
        Asserts.nonNull(topicQueryVo, RhinInfraErrorType.SUBSCRIBE_TOPIC_QUERY_IS_NULL);
        List<TopicQueryFilterVO> filterVoList = topicQueryHandler.queryTopicQueryFilterByExp(topicExp, queryExp);
        List<String> filterExps = filterVoList.stream().map(TopicQueryFilterVO::getFilterExp).collect(Collectors.toList());
        boolean containsFilterExp = false;
        for (SubscribeTopicFilterVO filterVo : filterList) {
            if (!filterExps.contains(filterVo.getQueryFilterExp())) {
                containsFilterExp = true;
                break;
            }
        }
        //判断订阅的过滤条件是否存在于主题查询函数下的过滤条件中
        Asserts.isFalse(containsFilterExp, RhinInfraErrorType.SUBSCRIBE_TOPIC_FILTER_NOT_EXIST);
    }


    /**
     * 订阅主题时校验订阅者是否存在
     *
     * @param subscriberCode
     */
    private void subscribeCheckSubscriber(String subscriberCode) {
        ParticipantDO participantDO = participantQueryHandler.queryByCodeAndRole(subscriberCode, EnumParticipantRole.SUBSCRIBER);
        Asserts.nonNull(participantDO, RhinInfraErrorType.SUBSCRIBE_TOPIC_SUBSCRIBER_IS_NULL);
    }

    /**
     * 发送mq消息
     *
     * @param publishMessageList
     */
    private void sendMessage(List<PublishMessageDto> publishMessageList) {
        String data = NotificationMessageDto.builder()
                .sender(serviceName)
                .data(publishMessageList)
                .build();
        rabbitTemplate.convertAndSend(rhinRabbitProperties.getExchange(), rhinRabbitProperties.getPublishMessageRoutingKey(), data);
    }
}
